package com.lj.queue.handler;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.lj.common.exception.CommonException;
import com.lj.queue.annotation.DelayQueueHandler;
import com.lj.queue.entity.DelayQueueMessage;
import com.lj.queue.entity.DelayQueueMeta;
import com.lj.queue.enums.DelayQueueMessageStatus;
import com.lj.queue.service.DelayQueueMessageService;
import com.lj.queue.service.DelayQueueMetaService;
import com.lj.queue.utils.DelayQueueUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.List;

/**
 * @author luojing
 * @since 2025/7/17 13:38
 */
@DelayQueueHandler(order = -1)
@Slf4j
public class PreloadHandler extends AbstractHandler<PreloadHandler.PreloadMessage> {
    @Override
    public PreloadMessage convertMessage(DelayHandlerContext context) throws Exception {
        PreloadMessage bean = super.convertMessage(context);
        if (bean == null || bean.get_start() == null) {
            throw new CommonException("不被支持的消息");
        }
        return bean;
    }

    @Override
    public void execute(DelayHandlerContext handlerContext, PreloadMessage message, DelayHandlerChain handlerChain) {
        LocalDateTime now = LocalDateTime.now();
        LocalDateTime nextTime = now;
        log.info("队列:[{}] 开始预加载消息...", handlerContext.getQueueName());
        final DelayQueueMetaService queueMetaService = handlerContext.getDelayQueueMetaService();
        final DelayQueueMessageService queueMessageService = handlerContext.getQueueMessageService();
        final TimeInterval timer = new TimeInterval();
        int preloadNum = 0;
        Integer version = message.get_version();
        try {
            // 先获取队列信息
            DelayQueueMeta queueMeta = queueMetaService.getById(handlerContext.getQueueId());
            if (queueMeta == null) {
                // 这里直接返回是防止删除了队列数据时还没来得及删除队列实例，会立刻触发删除实例
                return;
            }
            ChronoUnit chronoUnit = queueMeta.getTimeUnit().getChronoUnit();
            // 计算下次执行时间
            nextTime = now.plus(queueMeta.getPreloadDistance(), chronoUnit);
            // 开始预加载
            LambdaQueryWrapper<DelayQueueMessage> queryWrapper = queueMessageService.lambdaQueryWrapper()
                    .select(DelayQueueMessage::getId, DelayQueueMessage::getQueueId, DelayQueueMessage::getMessage, DelayQueueMessage::getExpectTime)
                    .eq(DelayQueueMessage::getQueueId, handlerContext.getQueueId());
            if (message.get_start()) {
                // 项目刚启动，加载之前加载过的和接下来待执行的消息
                queryWrapper.in(DelayQueueMessage::getStatus, DelayQueueMessageStatus.loaded, DelayQueueMessageStatus.pending)
                        .between(DelayQueueMessage::getExpectTime, queueMeta.getLastLoadTime(), now.plus(queueMeta.getPreloadDuration(), chronoUnit));
            } else {
                LocalDateTime startTime = now;
                if (queueMeta.getVersion() > version) {
                    // 队列信息版本有更新，防止出现消息的丢失，从上次加载时间开始
                    startTime = queueMeta.getLastLoadTime();
                    version = queueMeta.getVersion();
                }
                // 正常执行
                queryWrapper.eq(DelayQueueMessage::getStatus, DelayQueueMessageStatus.pending)
                        .between(DelayQueueMessage::getExpectTime, startTime, now.plus(queueMeta.getPreloadDuration(), chronoUnit));
            }
            List<DelayQueueMessage> queueMessageList = queueMessageService.list(queryWrapper);
            if (CollUtil.isNotEmpty(queueMessageList)) {
                preloadNum = queueMessageList.size();
                //放到队列
                DelayQueueUtils.pushQueue(handlerContext.getDelayQueue(), queueMessageList);
                // 更新消息状态
                queueMessageService.updateStatus(queueMessageList.stream().map(DelayQueueMessage::getId).toList(), DelayQueueMessageStatus.loaded);
            }
            // 更新最后预加载时间
            queueMetaService.updateLastLoadTime(queueMeta.getId(), now);
        } finally {
            // 不管是否出现了异常，都要为下次预加载做准备
            DelayQueueUtils.pushQueue(handlerContext.getDelayQueue(), DelayedMessage.of(nextTime)
                    .setQueueId(handlerContext.getQueueId())
                    .setMessage(JSONUtil.toJsonStr(new PreloadHandler.PreloadMessage(false, version))));
            log.info("队列:[{}] 预加载消息结束! 耗时: {}ms,预加载消息({})条,下次预加载时间: {}",
                    handlerContext.getQueueName(), timer.intervalMs(), preloadNum, nextTime);
        }
    }

    @Data
    public static class PreloadMessage {
        // 本次是否是刚启动
        private Boolean _start;

        // 队列信息版本
        private Integer _version;

        public PreloadMessage(Boolean _start, Integer _version) {
            this._start = _start;
            this._version = _version;
        }
    }
}
